package cn.tedu.charging.order.mqtt;

import com.alibaba.fastjson2.JSON;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;

/**
 * 是当前业务中 对所有消息发送数据的封装
 * 既可以发送订单消息 也可以发送计价消息 还可以发送 充电消息
 */
@Component
@Slf4j
public class MqttProducer<T> {
    //生产者发送的消息 执行者mqttClient
    @Autowired
    private MqttClient mqttClient;
    //给业务层提供一个生产发送消息的方法
    //1.发送到目标 destination
    //2.发送的消息数据 T
    //3.反馈发送结果 成功还是失败 boolean

    public boolean send(String destination,T msg){
        log.debug("开始向topic:{},发送消息:{}",destination,msg);
        //消息序列化问题 msg序列化成byte[] T-->String json-->byte[]
        byte[] payLoad= serialize(msg);
        //向destination发送 payload qos选择级别是1 并且要求主题保存消息(true 有一个新的消费者 重新获取所有消息 持久化)
        //qos=1可能重试发送 消息retain保存 每当启动一个消费者监听这个主题都会重新消费持久化消息
        //非常容易造成重复消费
        try{
            mqttClient.publish(destination,payLoad,1,true);
        }catch (Exception e){
            log.error("向topic:{},发送消息:{} 失败",destination,msg);
            return false;
        }
        return true;
    }

    private byte[] serialize(T msg) {
        //转化json
        String jsonString = JSON.toJSONString(msg);
        return jsonString.getBytes(StandardCharsets.UTF_8);
    }
}
